/****************************************************
 * 创建人:   robin
 * 创建时间: 2023/8/30.030 14:39
 * 项目名称: cpy
 * 文件名称: HandleMessageServiceImpl.java
 * 文件描述:
 *
 * All rights Reserved, Designed By 德品软件
 * @Copyright:2023-2023
 *
 ********************************************************/
package com.dp.service.impl;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.dp.config.ServiceConfig;
import com.dp.constant.CollectConstant;
import com.dp.constant.RocketMqSysConstant;
import com.dp.dto.*;
import com.dp.enumeration.ReturnCodeEnum;
import com.dp.enumeration.ServerPortEnum;
import com.dp.enumeration.ServerTypeEnum;
import com.dp.feign.PsdpFeign;
import com.dp.feign.vo.request.AddTradRecord4ReqVO;
import com.dp.feign.vo.request.EditDtuInfoReqVO;
import com.dp.feign.vo.request.EditOilTankReqVO;
import com.dp.feign.vo.request.LoginReqVO;
import com.dp.feign.vo.response.GetDtuInfoByNameRspVO;
import com.dp.feign.vo.response.LoginRspVO;
import com.dp.feign.vo.response.PsdpResponseData;
import com.dp.service.HandleMessageService;
import com.dp.template.RocketMqTemplate;
import com.dp.utils.BufferUtil;
import com.dp.utils.ObjectUtil;
import com.dp.utils.OtherUtil;
import com.dp.utils.TaskThreadPool;
import com.dp.vo.response.ResponseData;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * 包名称：com.dp.service.impl
 * 类名称：HandleMessageServiceImpl
 * 类描述：
 * 创建人： robin
 * 创建时间：2023/8/30.030 14:39
 */
@Slf4j
@Service
public class HandleMessageServiceImpl implements HandleMessageService {

    @Autowired
    private PsdpFeign psdpFeign;

    @Resource
    private RocketMqTemplate rocketMqTemplate;

    @Autowired
    private ServiceConfig serviceConfig;

    private Map<String, String> oilMap = new HashMap<>();
    private Map<String, String> oilTankMap = new HashMap<>();
    private Map<Integer, MsgCacheUnitDto> msgCacheUnitMap = new HashMap<>();
    private String psdpToken;
    private LoginReqVO psdpLoginReqVO;
    private BlockingQueue mqQueue;

    ByteBuffer tankHeartbeatByteBuffer;
    ByteBuffer tankDataCmdByteBuffer;
    ByteBuffer oilRegisterTipByteBuffer;
    ByteBuffer askByteBuffer;

    @Override
    public ResponseData initService() {
        try {

            psdpToken = "";
            psdpLoginReqVO = new LoginReqVO();
            psdpLoginReqVO.setUsername(serviceConfig.getUsername());
            psdpLoginReqVO.setPassword(serviceConfig.getPassword());

            for (ServerPortEnum serverPortEnum : ServerPortEnum.values()) {
                msgCacheUnitMap.put(serverPortEnum.getCode(), new MsgCacheUnitDto(serverPortEnum));
            }

            mqQueue = new LinkedBlockingQueue<MqDto>(CollectConstant.QUEUE_CAPACITY);

            tankHeartbeatByteBuffer = BufferUtil.hexStringToByteBuffer(CollectConstant.TANK_HEARTBEAT);
            tankDataCmdByteBuffer = BufferUtil.hexStringToByteBuffer(CollectConstant.TANK_DATA_CMD);
            oilRegisterTipByteBuffer = ByteBuffer.allocate(CollectConstant.OIL_REGISTER_TIP.length()).order(ByteOrder.BIG_ENDIAN).put(CollectConstant.OIL_REGISTER_TIP.getBytes());
            askByteBuffer = ByteBuffer.allocate(CollectConstant.ASK.length()).order(ByteOrder.BIG_ENDIAN).put(CollectConstant.ASK.getBytes());

            return ResponseData.handleSuccess("初始化处理消息业务成功");
        } catch (Exception e) {
            log.error(e.getMessage());
            return ResponseData.handleError("初始化处理消息业务失败！" + e.getMessage());
        }
    }

    @Override
    public ResponseData pushRecvData(Integer serverPort, MsgUnitDto msgUnitDto) {
        try {
            MsgCacheUnitDto msgCacheUnitDto = msgCacheUnitMap.get(serverPort);
            if (null == msgCacheUnitDto) {
                return ResponseData.handleError(ReturnCodeEnum.DEVICE_FIND_DEVICE_COMM_CTRL_OBJ_FAILED, serverPort);
            }
            if (!msgCacheUnitDto.getRecvQueue().offer(msgUnitDto)) {
                return ResponseData.handleWarn(ReturnCodeEnum.DEVICE_PUSH_RECV_MSG_QUEUE_FULL, serverPort);
            }
            return ResponseData.handleSuccess("成功");
        } catch (Exception e) {
            return ResponseData.handleError(ReturnCodeEnum.DEVICE_PUSH_RECV_MSG_QUEUE_EXCEPTION, serverPort, e.getMessage());
        }
    }

    @Override
    public ResponseData<MsgUnitDto> pollRecvData(Integer serverPort) {
        try {
            MsgCacheUnitDto msgCacheUnitDto = msgCacheUnitMap.get(serverPort);
            if (null == msgCacheUnitDto) {
                return ResponseData.handleError(ReturnCodeEnum.DEVICE_FIND_DEVICE_COMM_CTRL_OBJ_FAILED, serverPort);
            }
            Object object = msgCacheUnitDto.getRecvQueue().poll();
            if (null == object) {
                return ResponseData.handleWarnYellow(ReturnCodeEnum.DEVICE_POLL_RECV_MSG_QUEUE_EMPTY, serverPort);
            }
            return ResponseData.handleSuccess("成功", (MsgUnitDto)object);
        } catch (Exception e) {
            return ResponseData.handleError(ReturnCodeEnum.DEVICE_POLL_RECV_MSG_QUEUE_EXCEPTION, serverPort, e.getMessage());
        }
    }

    @Override
    public ResponseData pushSendData(Integer serverPort, String key, ByteBuffer data) {
        try {
            MsgCacheUnitDto msgCacheUnitDto = msgCacheUnitMap.get(serverPort);
            if (null == msgCacheUnitDto) {
                return ResponseData.handleError(ReturnCodeEnum.DEVICE_FIND_DEVICE_COMM_CTRL_OBJ_FAILED, serverPort);
            }
            BlockingQueue snedQueue = msgCacheUnitDto.getSendMap().get(key);
            if (null == snedQueue) {
                snedQueue = new LinkedBlockingQueue<ByteBuffer>(CollectConstant.QUEUE_CAPACITY);
                msgCacheUnitDto.getSendMap().put(key, snedQueue);
            }
            if (!snedQueue.offer(data)) {
                return ResponseData.handleWarn(ReturnCodeEnum.DEVICE_PUSH_SEND_MSG_QUEUE_FULL, serverPort, key);
            }
            return ResponseData.handleSuccess("成功");
        } catch (Exception e) {
            return ResponseData.handleError(ReturnCodeEnum.DEVICE_PUSH_SEND_MSG_QUEUE_EXCEPTION, serverPort, e.getMessage());
        }
    }

    @Override
    public ResponseData<ByteBuffer> pollSendData(Integer serverPort, String key) {
        try {
            MsgCacheUnitDto msgCacheUnitDto = msgCacheUnitMap.get(serverPort);
            if (null == msgCacheUnitDto) {
                return ResponseData.handleError(ReturnCodeEnum.DEVICE_FIND_DEVICE_COMM_CTRL_OBJ_FAILED, serverPort);
            }
            BlockingQueue snedQueue = msgCacheUnitDto.getSendMap().get(key);
            if (null == snedQueue) {
                return ResponseData.handleWarnYellow(ReturnCodeEnum.DEVICE_POLL_SEND_MSG_QUEUE_EMPTY, serverPort);
            }
            Object object = snedQueue.poll();
            if (null == object) {
                return ResponseData.handleWarnYellow(ReturnCodeEnum.DEVICE_POLL_SEND_MSG_QUEUE_EMPTY, serverPort);
            }
            return ResponseData.handleSuccess("成功", (ByteBuffer)object);
        } catch (Exception e) {
            return ResponseData.handleError(ReturnCodeEnum.DEVICE_POLL_SEND_MSG_QUEUE_EXCEPTION, serverPort, e.getMessage());
        }
    }

    @Override
    public ResponseData pushMq(String tag, BaseMqMessage msg) {
        try {
            MqDto mqDto = new MqDto();
            mqDto.setDestination(CollectConstant.TOPIC + RocketMqSysConstant.DELIMITER + tag);
            mqDto.setMsg(msg);
            if (!mqQueue.offer(mqDto)) {
                return ResponseData.handleWarn(ReturnCodeEnum.DEVICE_PUSH_MQ_QUEUE_FULL);
            }
            return ResponseData.handleSuccess("成功");
        } catch (Exception e) {
            return ResponseData.handleError(ReturnCodeEnum.DEVICE_PUSH_MQ_QUEUE_EXCEPTION, e.getMessage());
        }
    }

    @Override
    public boolean handleRecvMsg() {
        AtomicBoolean noMsg = new AtomicBoolean(true);

        try {
            msgCacheUnitMap.values().forEach(msgCacheUnitDto -> {
                Object object = null;
                while ((object = msgCacheUnitDto.getRecvQueue().poll()) != null) {
                    noMsg.set(false);
                    parseRecvMsg((MsgUnitDto)object);
                }
            });

        } catch (Exception e) {
            log.error("handleRecvMsg Exception:", e);
        }

        return noMsg.get();
    }

    private void parseRecvMsg(MsgUnitDto msgUnitDto) {
        switch (msgUnitDto.getServerPortEnum().getType()) {
            case NOZZLE:
                List<MsgUnitDto> msgUnitDtoList = splitNozzleMsg(msgUnitDto);
                log.debug("msgUnitDtoList size:{}", msgUnitDtoList.size());
                for (MsgUnitDto msg : msgUnitDtoList) {
                    parseNozzleMsg(msg);
                }
                break;

            case OIL_TANK:
                parseTankMsg(msgUnitDto);
                break;

            default:
                break;
        }
    }

    private List<MsgUnitDto> splitNozzleMsg(MsgUnitDto msgUnitDto) {
        List<MsgUnitDto> msgUnitDtoList = new ArrayList<>();
        short head = 0;
        short length = 0;
        ByteBuffer byteBuffer = msgUnitDto.getByteBuffer();
        String str = StandardCharsets.UTF_8.decode(byteBuffer).toString();
        while (byteBuffer.position() < byteBuffer.limit()) {
            log.debug("begin byteBuffer position{} limit{}", byteBuffer.position(), byteBuffer.limit());

            head = BufferUtil.getShort(byteBuffer);
            if (head != CollectConstant.OIL_DATA_HEAD) {
                byteBuffer.rewind();
                log.error("parseRecvMsg failed,message header not found:{}", BufferUtil.byteBufferToHexString(byteBuffer, byteBuffer.limit(), true));
                break;
            }
            length = BufferUtil.getShort(byteBuffer);
            MsgUnitDto newMsgUnitDto = new MsgUnitDto();
            newMsgUnitDto.setServerPortEnum(msgUnitDto.getServerPortEnum());
            newMsgUnitDto.setEquipmentIp(msgUnitDto.getEquipmentIp());
            newMsgUnitDto.setPort(msgUnitDto.getPort());

            ByteBuffer newByteBuffer = ByteBuffer.allocate(length + CollectConstant.MSG_HEAD_LENGTH).order(ByteOrder.BIG_ENDIAN);
            newByteBuffer.putShort(head);
            newByteBuffer.putShort(length);
            BufferUtil.copyByteBuffer(byteBuffer, newByteBuffer, length);
            newByteBuffer.flip();
            newMsgUnitDto.setByteBuffer(newByteBuffer);
            msgUnitDtoList.add(newMsgUnitDto);

            log.debug("end byteBuffer position{} limit{}", byteBuffer.position(), byteBuffer.limit());
        }
        return msgUnitDtoList;
    }

    private void parseNozzleMsg(MsgUnitDto msgUnitDto) {
        DateTime dateTime = DateUtil.date();
        short head = 0;
        short length = 0;
        short cmd = 0;
        String gatewayNo = null;
        String dtuInfoId = null;
        EncoderOilRecordDto encoderOilRecordDto = null;
        NozzleUseRecordDto nozzleUseRecordDto = null;
        TaxDeclarationRecordDto taxDeclarationRecordDto = null;
        ByteBuffer byteBuffer = msgUnitDto.getByteBuffer();
        do {
            head = BufferUtil.getShort(byteBuffer);
            if (head != CollectConstant.OIL_DATA_HEAD) {
                byteBuffer.rewind();
                log.error("parseRecvMsg failed,message header not found:{}", BufferUtil.byteBufferToHexString(byteBuffer, byteBuffer.limit(), true));
                break;
            }
            length = BufferUtil.getShort(byteBuffer); if (Short.MIN_VALUE == length) break;
            cmd = BufferUtil.getShort(byteBuffer); if (Short.MIN_VALUE == cmd) break;
            gatewayNo = BufferUtil.byteBufferToHexString(byteBuffer, CollectConstant.OIL_GATEWAY_LENGTH, false); if (StringUtils.isBlank(gatewayNo)) break;
            if (!BufferUtil.skip(byteBuffer, 1)) break;

            if (serviceConfig.isPsdpSwitch()) {
                dtuInfoId = oilMap.get(gatewayNo);
                if (null == dtuInfoId) {
                    PsdpResponseData<GetDtuInfoByNameRspVO> rsp = psdpFeign.getDtuInfoByName(gatewayNo, psdpToken);
                    if (!rsp.getSuccess()) {
                        log.warn("getDtuInfoByName[{}] return:{}-{}", gatewayNo, rsp.getCode(), rsp.getMessage());
                        pushSendData(msgUnitDto.getServerPortEnum().getCode(), OtherUtil.getIpPortString(msgUnitDto.getEquipmentIp(), msgUnitDto.getPort()), BufferUtil.cloneByteBuffer(oilRegisterTipByteBuffer));

                    } else {
                        dtuInfoId = rsp.getResult().getId();
                        oilMap.put(gatewayNo, dtuInfoId);
                        log.debug("[{}] register was successful, dtuInfoId:{}", gatewayNo, dtuInfoId);
                    }

                }
            }

            switch (cmd) {
                case CollectConstant.OIL_DATA_CMD_OIL:
                case CollectConstant.OIL_DATA_CMD_OIL_2:
                    int packenum = BufferUtil.getInt(byteBuffer); if (Integer.MIN_VALUE == packenum) break;
                    int timestamp = BufferUtil.getInt(byteBuffer); if (Integer.MIN_VALUE == timestamp) break;
                    int oilVolume = BufferUtil.getInt(byteBuffer); if (Integer.MIN_VALUE == oilVolume) break;
                    int oilDuration = BufferUtil.getInt(byteBuffer); if (Integer.MIN_VALUE == oilDuration) break;

                    encoderOilRecordDto = handleEncoder(gatewayNo, msgUnitDto.getServerPortEnum(), dateTime, oilVolume);
                    if (encoderOilRecordDto != null && encoderOilRecordDto.getOilVolume().compareTo(CollectConstant.MAX_OIL_VOL) > 0) {
                        return;
                    }

                    if (!BufferUtil.skip(byteBuffer, 1)) break;

                    nozzleUseRecordDto = handleNozzle(encoderOilRecordDto.getSn(), byteBuffer, gatewayNo, msgUnitDto.getServerPortEnum(), dateTime, oilDuration);
                    taxDeclarationRecordDto = handleTax(encoderOilRecordDto.getSn(), byteBuffer, gatewayNo, msgUnitDto.getServerPortEnum(), dateTime);
                    if (taxDeclarationRecordDto != null && taxDeclarationRecordDto.getOilVolume().compareTo(CollectConstant.MAX_OIL_VOL) > 0) {
                        return;
                    }

                    addTradRecord4Psdp(byteBuffer, length, msgUnitDto, gatewayNo, dateTime, encoderOilRecordDto, nozzleUseRecordDto, taxDeclarationRecordDto);
                    break;

                case CollectConstant.OIL_DATA_CMD_HEARTBEAT:
                case CollectConstant.OIL_DATA_CMD_HEARTBEAT_2:
                    handleHeartbeat(gatewayNo, dateTime);
                    editDtuInfoPsdp(dtuInfoId, dateTime);
                    break;

                case CollectConstant.OIL_DATA_CMD_LIFT_TIMEOUT:
                    log.debug("lift timeout data");
                    break;

                case CollectConstant.UNKNOWN_CMD_01:
                    log.debug("Unknown command: 0x{}", BufferUtil.bytesToHexString(BufferUtil.shortToBytes(cmd), 2, false));
                    break;

                default:
                    log.info("Unknown command: 0x{}", BufferUtil.bytesToHexString(BufferUtil.shortToBytes(cmd), 2, false));
                    break;
            }
        } while (false);
        sendOilRecord(encoderOilRecordDto, taxDeclarationRecordDto, nozzleUseRecordDto);
    }

    private void parseTankMsg(MsgUnitDto msgUnitDto) {
        DateTime dateTime = DateUtil.date();
        short head = 0;
        short length = 0;
        short cmd = 0;
        String gatewayNo = null;
        ByteBuffer byteBuffer = msgUnitDto.getByteBuffer();
        do {
            head = BufferUtil.getShort(byteBuffer);
            if (CollectConstant.TANK_DATA_HEAD == head) {
                length = BufferUtil.getShort(byteBuffer);
                if (CollectConstant.TANK_DATA_LENGTH == length) {
                    cmd = BufferUtil.getShort(byteBuffer);
                    if (CollectConstant.TANK_DATA_CMD_HEARTBEAT == cmd) {
                        gatewayNo = CollectConstant.TANK_GATEWAY_PREFIX + BufferUtil.byteBufferToString(byteBuffer, CollectConstant.TANK_GATEWAY_LENGTH);
                        oilTankMap.put(OtherUtil.getIpPortString(msgUnitDto.getEquipmentIp(), msgUnitDto.getPort()), gatewayNo);
                        handleHeartbeat(gatewayNo, dateTime);
                        editOilTankPsdp(null, gatewayNo, null);
                    }
                }
            } else if (0x0169 == head) {
                byteBuffer.rewind();
                String tankDataCmd = BufferUtil.byteBufferToHexString(byteBuffer, 7, false);
                if (!CollectConstant.TANK_DATA_CMD.equals(tankDataCmd)) {
                    log.info("Unknown command:Expecting 0x {}", CollectConstant.TANK_DATA_CMD);
                    break;
                }
                gatewayNo = oilTankMap.get(OtherUtil.getIpPortString(msgUnitDto.getEquipmentIp(), msgUnitDto.getPort()));
                if (StringUtils.isBlank(gatewayNo)) {
                    log.info("Failed to find gateway device number through IP[{}:{}]", msgUnitDto.getEquipmentIp(), msgUnitDto.getPort());
                    break;
                }
                handleTank(byteBuffer, gatewayNo, dateTime);
            }
            break;
        } while (false);
    }

    private void sendOilRecord(EncoderOilRecordDto encoderOilRecordDto, TaxDeclarationRecordDto taxDeclarationRecordDto, NozzleUseRecordDto nozzleUseRecordDto) {
        if (encoderOilRecordDto != null) {
            if (taxDeclarationRecordDto != null && taxDeclarationRecordDto.getPrice().compareTo(BigDecimal.ZERO) > 0) {
                encoderOilRecordDto.setPrice(taxDeclarationRecordDto.getPrice());
                encoderOilRecordDto.setAmount(taxDeclarationRecordDto.getPrice().multiply(encoderOilRecordDto.getOilVolume()).setScale(2, BigDecimal.ROUND_HALF_UP));
            }
            pushMq(CollectConstant.IM_ENCODER_OIL_RECORD_TAG, encoderOilRecordDto);
        }
        if (taxDeclarationRecordDto != null) {
            pushMq(CollectConstant.IM_TAX_DECLARATION_RECORD_TAG, taxDeclarationRecordDto);
        }
        if (nozzleUseRecordDto != null) {
            pushMq(CollectConstant.IM_NOZZLE_USE_RECORD_TAG, nozzleUseRecordDto);
        }
    }

    private void handleTank(ByteBuffer byteBuffer, String gatewayNo, DateTime dateTime) {
        List<OilTankStatusDto> oilTankStatusDtoList = new ArrayList<>();
        BufferUtil.skip(byteBuffer, 19);
        String bufferString = BufferUtil.byteBufferToString(byteBuffer, byteBuffer.limit() - 26);
        int index = bufferString.indexOf("&&");
        if (index < 0) {
            return;
        }
        String substring = bufferString.substring(0, index);
        String[] tanks = substring.split("07");
        for (int i = 0; i < tanks.length; i++) {
            if (tanks[i].length() >= 48) {
                OilTankStatusDto oilTankStatusDto = new OilTankStatusDto();
                oilTankStatusDto.setSn(IdWorker.getId());
                oilTankStatusDto.setGatewayEquipmentNo(gatewayNo);
                oilTankStatusDto.setOilTankSn(i+1);
                BigDecimal bigDecimal = BufferUtil.hexStringToBigDecimal(tanks[i].substring(0, 8));
                oilTankStatusDto.setOilVolume(bigDecimal.compareTo(BigDecimal.ZERO) >= 0 ? bigDecimal : BigDecimal.ZERO);
                bigDecimal = BufferUtil.hexStringToBigDecimal(tanks[i].substring(24, 32));
                oilTankStatusDto.setOilHight(bigDecimal.compareTo(BigDecimal.ZERO) >= 0 ? bigDecimal : BigDecimal.ZERO);
                bigDecimal = BufferUtil.hexStringToBigDecimal(tanks[i].substring(40, 48));
                oilTankStatusDto.setTemperature(bigDecimal.compareTo(BigDecimal.ZERO) >= 0 ? bigDecimal : BigDecimal.ZERO);
                oilTankStatusDto.setCollectionTime(dateTime.toString(CollectConstant.FORMAT_DATE_YMDHMS));
                log.debug("oilTankStatusDto:{}", JSONObject.toJSONString(oilTankStatusDto));

                pushMq(CollectConstant.IM_OIL_TANK_STATUS_TAG, oilTankStatusDto);

                oilTankStatusDtoList.add(oilTankStatusDto);
            }
        }
        editOilTankPsdp(byteBuffer, gatewayNo, oilTankStatusDtoList);
    }

    private EncoderOilRecordDto handleEncoder(String gatewayNo, ServerPortEnum serverPortEnum, DateTime dateTime, int oilVolume) {
        EncoderOilRecordDto encoderOilRecordDto = new EncoderOilRecordDto();
        encoderOilRecordDto.setSn(IdWorker.getId());
        encoderOilRecordDto.setGatewayEquipmentNo(gatewayNo);
        encoderOilRecordDto.setPort(serverPortEnum.getCode());
        encoderOilRecordDto.setOilVolume(BigDecimal.valueOf(oilVolume).divide(CollectConstant.BIGDECIMAL_HUNDRED, 2, BigDecimal.ROUND_HALF_UP));
        encoderOilRecordDto.setPrice(BigDecimal.ZERO);
        encoderOilRecordDto.setAmount(BigDecimal.ZERO);
        encoderOilRecordDto.setRecordTime(dateTime.toString(CollectConstant.FORMAT_DATE_YMDHMS));
        log.debug("encoderOilRecordDto:{}", JSONObject.toJSONString(encoderOilRecordDto));

        return encoderOilRecordDto;
    }

    private NozzleUseRecordDto handleNozzle(Long sn, ByteBuffer byteBuffer, String gatewayNo, ServerPortEnum serverPortEnum, DateTime dateTime, int oilDuration) {
        try {
            String hour = BufferUtil.byteBufferToAsciiString(byteBuffer, 2);
            if (StringUtils.isBlank(hour)) {
                log.error("handleNozzle get hour failed");
                return null;
            }
            String minute = BufferUtil.byteBufferToAsciiString(byteBuffer, 2);
            if (StringUtils.isBlank(minute)) {
                log.error("handleNozzle get minute failed");
                return null;
            }
            String second = BufferUtil.byteBufferToAsciiString(byteBuffer, 2);
            if (StringUtils.isBlank(second)) {
                log.error("handleNozzle get second failed");
                return null;
            }
            DateTime liftTime = DateUtil.parse(dateTime.toString(CollectConstant.FORMAT_DATE_YMD) + CollectConstant.SPACE + hour + CollectConstant.COLON + minute + CollectConstant.COLON + second, CollectConstant.FORMAT_DATE_YMDHMS);

            NozzleUseRecordDto nozzleUseRecordDto = new NozzleUseRecordDto();
            nozzleUseRecordDto.setSn(sn);
            nozzleUseRecordDto.setGatewayEquipmentNo(gatewayNo);
            nozzleUseRecordDto.setPort(serverPortEnum.getCode());
            nozzleUseRecordDto.setDuration(oilDuration);
            nozzleUseRecordDto.setLiftTime(liftTime.toString(CollectConstant.FORMAT_DATE_YMDHMS));
            nozzleUseRecordDto.setHangUpTime(DateUtil.offsetSecond(liftTime, oilDuration).toString(CollectConstant.FORMAT_DATE_YMDHMS));
            nozzleUseRecordDto.setRecordTime(dateTime.toString(CollectConstant.FORMAT_DATE_YMDHMS));
            log.debug("nozzleUseRecordDto:{}", JSONObject.toJSONString(nozzleUseRecordDto));

            return nozzleUseRecordDto;
        } catch (Exception e) {
            log.error("handleNozzle Exception:", e);
            return null;
        }
    }

    private TaxDeclarationRecordDto handleTax(Long sn, ByteBuffer byteBuffer, String gatewayNo, ServerPortEnum serverPortEnum, DateTime dateTime) {
        do {
            if (!BufferUtil.skip(byteBuffer, 1)) break;
            String oilVolumeString = BufferUtil.byteBufferToAsciiString(byteBuffer, 6);
            if (StringUtils.isBlank(oilVolumeString)) {
                log.debug("handleTax get oilVolumeString failed");
                break;
            }
            if (!BufferUtil.skip(byteBuffer, 1)) break;
            String amountString = BufferUtil.byteBufferToAsciiString(byteBuffer, 6);
            if (StringUtils.isBlank(amountString)) {
                log.debug("handleTax get amountString failed");
                break;
            }
            if (!BufferUtil.skip(byteBuffer, 1)) break;
            String priceString = BufferUtil.byteBufferToAsciiString(byteBuffer, 4);
            if (StringUtils.isBlank(priceString)) {
                log.debug("handleTax get priceString failed");
                break;
            }

            try {
                TaxDeclarationRecordDto taxDeclarationRecordDto = new TaxDeclarationRecordDto();
                taxDeclarationRecordDto.setSn(sn);
                taxDeclarationRecordDto.setGatewayEquipmentNo(gatewayNo);
                taxDeclarationRecordDto.setPort(serverPortEnum.getCode());
                taxDeclarationRecordDto.setRecordTime(dateTime.toString(CollectConstant.FORMAT_DATE_YMDHMS));
                taxDeclarationRecordDto.setOilVolume(BigDecimal.valueOf(Integer.parseInt(oilVolumeString)).divide(CollectConstant.BIGDECIMAL_HUNDRED, 2, BigDecimal.ROUND_HALF_UP));
                taxDeclarationRecordDto.setAmount(BigDecimal.valueOf(Integer.parseInt(amountString)).divide(CollectConstant.BIGDECIMAL_HUNDRED, 2, BigDecimal.ROUND_HALF_UP));
                taxDeclarationRecordDto.setPrice(BigDecimal.valueOf(Integer.parseInt(priceString)).divide(CollectConstant.BIGDECIMAL_HUNDRED, 2, BigDecimal.ROUND_HALF_UP));
                log.debug("handleTax:taxDeclarationRecordDto:{}", JSONObject.toJSONString(taxDeclarationRecordDto));
                return taxDeclarationRecordDto;
            } catch (Exception e) {
                log.error("handleTax Exception:", e);
            }
        } while (false);
        return null;
    }

    private TaxDeclarationRecordDto handleTaxEx(TaxDeclarationRecordDto taxDeclarationRecordDto, ByteBuffer byteBuffer) {
        try {
            byteBuffer.rewind();
            String hexString = BufferUtil.byteBufferToHexString(byteBuffer, byteBuffer.limit(), false);
            if (StringUtils.isBlank(hexString)) {
                return null;
            }
            int index = hexString.indexOf(CollectConstant.OIL_DATA_AMOUNT_1);
            if (index < 0) {
                index = hexString.indexOf(CollectConstant.OIL_DATA_AMOUNT_2);
            }
            if (index < 0) {
                return null;
            }
            taxDeclarationRecordDto.setAmount(BufferUtil.doubleHexStringToBigDecimal(hexString.substring(index + 4, byteBuffer.limit() * 2 - 4)).setScale(2, BigDecimal.ROUND_HALF_UP));
            taxDeclarationRecordDto.setOilVolume(BigDecimal.ZERO);
            taxDeclarationRecordDto.setPrice(BigDecimal.ZERO);
            log.debug("handleTaxEx:taxDeclarationRecordDto:{}", JSONObject.toJSONString(taxDeclarationRecordDto));
            return taxDeclarationRecordDto;
        } catch (Exception e) {
            log.error("handleTaxEx Exception:", e);
        }
        return null;
    }

    private void handleHeartbeat(String gatewayNo, DateTime dateTime) {
        EquipmentHeartbeatDto equipmentHeartbeatDto = new EquipmentHeartbeatDto();
        equipmentHeartbeatDto.setSn(IdWorker.getId());
        equipmentHeartbeatDto.setEquipmentNo(gatewayNo);
        equipmentHeartbeatDto.setLastCollectTime(dateTime.toString(CollectConstant.FORMAT_DATE_YMDHMS));
        log.debug("equipmentHeartbeatDto:{}", JSONObject.toJSONString(equipmentHeartbeatDto));

        pushMq(CollectConstant.IM_EQUIPMENT_STATUS_TAG, equipmentHeartbeatDto);
    }

    @Override
    public void handlePsdpToken() {
        try {
            PsdpResponseData<LoginRspVO> rsp = psdpFeign.login(psdpLoginReqVO);
            if (200 == rsp.getCode()) {
                psdpToken = rsp.getResult().getToken();
                log.debug("handlePsdpToken:{}", psdpToken);
            } else {
                log.warn("handlePsdpToken return:{}-{}", rsp.getCode(), rsp.getMessage());
            }
        } catch (Exception e) {
            log.error("handlePsdpToken Exception:", e);
        }
    }

    private void editOilTankPsdp(ByteBuffer byteBuffer, String gatewayNo, List<OilTankStatusDto> oilTankStatusDtoList) {
        try {
            if (!serviceConfig.isPsdpSwitch()) {
                return;
            }

            log.debug("oilTankStatusDtoList:{}", JSONObject.toJSONString(oilTankStatusDtoList));

            EditOilTankReqVO editOilTankReqVO = new EditOilTankReqVO();
            editOilTankReqVO.setDtuNum(gatewayNo);
            if (CollectionUtils.isNotEmpty(oilTankStatusDtoList)) {
                for (OilTankStatusDto oilTankStatusDto : oilTankStatusDtoList) {
                    ObjectUtil.setObjectAttributeValue(editOilTankReqVO, String.format("tankSize%d", oilTankStatusDto.getOilTankSn()), oilTankStatusDto.getOilVolume());
                    ObjectUtil.setObjectAttributeValue(editOilTankReqVO, String.format("tankHeight%d", oilTankStatusDto.getOilTankSn()), oilTankStatusDto.getOilHight());
                    ObjectUtil.setObjectAttributeValue(editOilTankReqVO, String.format("temperature%d", oilTankStatusDto.getOilTankSn()), oilTankStatusDto.getTemperature());
                }
            }
            if (byteBuffer != null) {
                byteBuffer.rewind();
                editOilTankReqVO.setHex(BufferUtil.byteBufferToHexString(byteBuffer, byteBuffer.limit(), false));
            }
            log.debug("editOilTankReqVO:{}", JSONObject.toJSONString(editOilTankReqVO));
            PsdpResponseData rsp = psdpFeign.editOilTank(editOilTankReqVO, psdpToken);
            if (200 == rsp.getCode()) {
                log.debug("editOilTankPsdp ok");
            } else {
                log.warn("editOilTankPsdp return:{}-{}", rsp.getCode(), rsp.getMessage());
            }
        } catch (Exception e) {
            log.error("editOilTankPsdp Exception:", e);
        }
    }

    private void editDtuInfoPsdp(String dtuInfoId, DateTime dateTime) {
        try {
            if (!serviceConfig.isPsdpSwitch() || StringUtils.isBlank(dtuInfoId)) {
                return;
            }

            EditDtuInfoReqVO editDtuInfoReqVO = new EditDtuInfoReqVO();
            editDtuInfoReqVO.setId(dtuInfoId);
            editDtuInfoReqVO.setHeartbeat(dateTime.toString(CollectConstant.FORMAT_DATE_YMDHMS));
            log.debug("editDtuInfoReqVO:{}", JSONObject.toJSONString(editDtuInfoReqVO));
            PsdpResponseData rsp = psdpFeign.editDtuInfo(editDtuInfoReqVO, psdpToken);
            if (rsp.getSuccess()) {
                log.debug("editDtuInfoPsdp ok");
            } else {
                log.warn("editDtuInfoPsdp dtuInfoId[{}] return:{}-{}", dtuInfoId, rsp.getCode(), rsp.getMessage());
            }
        } catch (Exception e) {
            log.error("editDtuInfoPsdp Exception:", e);
        }
    }

    private void addTradRecord4Psdp(ByteBuffer byteBuffer,
                                    short length,
                                    MsgUnitDto msgUnitDto,
                                    String gatewayNo,
                                    DateTime dateTime,
                                    EncoderOilRecordDto encoderOilRecordDto,
                                    NozzleUseRecordDto nozzleUseRecordDto,
                                    TaxDeclarationRecordDto taxDeclarationRecordDto) {
        try {
            if (!serviceConfig.isPsdpSwitch()) {
                return;
            }

            AddTradRecord4ReqVO addTradRecord4ReqVO = new AddTradRecord4ReqVO();
            addTradRecord4ReqVO.setBianma(encoderOilRecordDto.getOilVolume().multiply(CollectConstant.BIGDECIMAL_HUNDRED).intValue());
            addTradRecord4ReqVO.setBuyDate(dateTime.toString(CollectConstant.FORMAT_DATE_YMDHMS));
            addTradRecord4ReqVO.setJiayoutime(nozzleUseRecordDto != null ? nozzleUseRecordDto.getDuration() : 0);
            addTradRecord4ReqVO.setOilCount(taxDeclarationRecordDto != null ? taxDeclarationRecordDto.getOilVolume() : BigDecimal.ZERO);
            addTradRecord4ReqVO.setTotalPrice(taxDeclarationRecordDto != null ? taxDeclarationRecordDto.getAmount() : BigDecimal.ZERO);
            addTradRecord4ReqVO.setUnitPrice(taxDeclarationRecordDto != null ? taxDeclarationRecordDto.getPrice() : BigDecimal.ZERO);
            addTradRecord4ReqVO.setIpaddr(msgUnitDto.getEquipmentIp());

            byteBuffer.rewind();
            addTradRecord4ReqVO.setHex(BufferUtil.byteBufferToHexString(byteBuffer, length + 4, false));
            log.debug("addTradRecord4ReqVO:{}", JSONObject.toJSONString(addTradRecord4ReqVO));

            PsdpResponseData rsp = null;
            if (addTradRecord4ReqVO.getTotalPrice().compareTo(BigDecimal.ZERO) > 0) {
                rsp = psdpFeign.addTradRecord("apiAdd3", addTradRecord4ReqVO, gatewayNo, msgUnitDto.getServerPortEnum().getCode(), psdpToken);
                log.debug("addTradRecord4Psdp apiAdd3");
            } else {
                rsp = psdpFeign.addTradRecord("apiAdd2", addTradRecord4ReqVO, gatewayNo, msgUnitDto.getServerPortEnum().getCode(), psdpToken);
                log.debug("addTradRecord4Psdp apiAdd2");
            }
            if (rsp.getSuccess()) {
                log.debug("addTradRecord4Psdp ok");
            } else {
                log.warn("addTradRecord4Psdp[{}-{}] return:{}-{}", gatewayNo, msgUnitDto.getServerPortEnum().getCode(), rsp.getCode(), rsp.getMessage());
            }
        } catch (Exception e) {
            log.error("addTradRecord4Psdp Exception:", e);
        }
    }

    @Override
    public ResponseData handleNewConnection(ServerPortEnum serverPortEnum, String ip, int port) {
        msgCacheUnitMap.get(serverPortEnum.getCode()).getSendMap().put(OtherUtil.getIpPortString(ip, port), new LinkedBlockingQueue<ByteBuffer>(CollectConstant.QUEUE_CAPACITY));
        if (serverPortEnum != null && StringUtils.isNotBlank(ip) && serverPortEnum.getType() == ServerTypeEnum.OIL_TANK) {
            pushSendData(serverPortEnum.getCode(), OtherUtil.getIpPortString(ip, port), BufferUtil.cloneByteBuffer(tankHeartbeatByteBuffer));
        }
        return ResponseData.handleSuccess("处理新连接成功");
    }

    @Override
    public ResponseData clearConnection(ServerPortEnum serverPortEnum, String ip, int port) {
        msgCacheUnitMap.get(serverPortEnum.getCode()).getSendMap().remove(OtherUtil.getIpPortString(ip, port));
        return ResponseData.handleSuccess("清理连接成功");
    }

    @Override
    public void sendAsk() {
//        msgCacheUnitMap.values().forEach(msgCacheUnitDto -> msgCacheUnitDto.getSendMap().keySet().forEach(ip -> pushSendData(msgCacheUnitDto.getServerPortEnum().getCode(), ip, BufferUtil.cloneByteBuffer(askByteBuffer))));
        Collection<MsgCacheUnitDto> msgCacheUnitDtos = msgCacheUnitMap.values();
        for (MsgCacheUnitDto msgCacheUnitDto : msgCacheUnitDtos) {
            int count = 0;
            Set<String> keySet = msgCacheUnitDto.getSendMap().keySet();
            for (String key : keySet) {
                pushSendData(msgCacheUnitDto.getServerPortEnum().getCode(), key, BufferUtil.cloneByteBuffer(askByteBuffer));
                count++;
            }
            log.info("sendAsk:[{}][{}]", msgCacheUnitDto.getServerPortEnum().getCode(), count);
        }

    }

    @Override
    public void sendTankHeartbeat() {
        MsgCacheUnitDto msgCacheUnitDto = msgCacheUnitMap.get(ServerPortEnum.PORT_OIL_TANK.getCode());
        if (null == msgCacheUnitDto) {
            return;
        }

        msgCacheUnitDto.getSendMap().keySet().forEach(key -> pushSendData(msgCacheUnitDto.getServerPortEnum().getCode(), key, BufferUtil.cloneByteBuffer(tankHeartbeatByteBuffer)));
    }

    @Override
    public void sendTankDataCmd() {
        MsgCacheUnitDto msgCacheUnitDto = msgCacheUnitMap.get(ServerPortEnum.PORT_OIL_TANK.getCode());
        if (null == msgCacheUnitDto) {
            return;
        }

        msgCacheUnitDto.getSendMap().keySet().forEach(key -> pushSendData(msgCacheUnitDto.getServerPortEnum().getCode(), key, BufferUtil.cloneByteBuffer(tankDataCmdByteBuffer)));
    }

    @Override
    public void sendMq() {
        try {
            Object object = null;
            while ((object = mqQueue.poll()) != null) {
                MqDto mqDto = (MqDto) object;
                rocketMqTemplate.send(mqDto.getDestination(), mqDto.getMsg());
            }
        } catch (Exception e) {
            log.error("sendMq Exception:", e);
        }

    }

    @Override
    public void printStatus() {
        try {
            TaskThreadPool.printStatus();
            msgCacheUnitMap.forEach((k, v) -> log.info("port:[{}],ip_size:[{}],key:[{}]", k, v.getSendMap().keySet().size(), v.getSendMap().keySet().stream().collect(Collectors.joining(CollectConstant.SEMICOLON))));
        } catch (Exception e) {
            log.error("printStatus Exception:", e);
        }
    }
}
